/*
 * Copyright 2011 The Rusted Roof, LLC.
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *        http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */

package org.proteusframework.platformservice.messaging;

import org.proteusframework.core.api.model.INamespace;
import org.proteusframework.core.base.AbstractObject;
import org.proteusframework.core.base.Namespace;
import org.proteusframework.core.util.Assert;
import org.proteusframework.platformservice.grammar.DefaultGrammarInspector;
import org.proteusframework.platformservice.grammar.IGrammarInspector;
import org.proteusframework.platformservice.grammar.IGrammarInspectorFactory;
import org.proteusframework.platformservice.grammar.IllegalExpressionException;
import org.proteusframework.platformservice.lifecycle.api.LifecycleActivityEvent;
import org.proteusframework.platformservice.lifecycle.api.LifecycleServiceIntentionEvent;
import org.proteusframework.platformservice.persistence.api.messagebean.IMessageBeanDescriptor;

import java.util.List;
import java.util.Properties;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Logger;

/**
 * Base class for all <code>IMessageSink</code> implementations.
 * <p/>
 * Model management inside of the base class is thread safe.
 *
 * @author Tacoma Four
 */
public abstract class AbstractMessageSink<T extends IMessageBean> extends AbstractObject implements IMessageSink<T>
{
    private static final Logger logger = Logger.getLogger(AbstractMessageSink.class.getCanonicalName());

    /**
     * Lock used to provide thread-safe access to the model management methods, i.e. <code>createSubscriber</code> and
     * <code>setMessageSinkListener</code>.
     */
    private final Object lock = new Object();

    /**
     * Collection of metadata describing the backing (physical) message service.
     */
    private final Properties metadata;

    /**
     * Message bean class that this sink is capable of managing.
     */
    private final Class<T> messageBeanClass;

    /**
     * Manages a list of sink subscribers.
     */
    protected List<T> subscribers = new CopyOnWriteArrayList<T>();

    /**
     * Manages a list of sink producers.
     */
    protected List<T> producers = new CopyOnWriteArrayList<T>();

    private boolean hasModelBeenDefined = false;

    private IGrammarInspectorFactory grammarInspectorFactory;

    /**
     * Flag used to track if the instance is relying upon a subscriber model or a sole registered listener model.
     */
    private boolean isSubscriberBasedModel = false;

    /**
     * Sole registered listener for the message sink.
     */
    protected IMessageSinkListener<T> registeredListener;

    protected AbstractMessageSink(INamespace namespace, Class<T> messageBeanClass, Properties metadata)
    {
        super(namespace);
        Assert.parameterNotNull(messageBeanClass, "Parameter 'messageBeanClass' must not be null");

        if (null == metadata)
        {
            this.metadata = new Properties();
        } else
        {
            this.metadata = metadata;
        }

        this.messageBeanClass = messageBeanClass;

        if (null != this.metadata.getProperty(IMessageService.KEY_SUBSCRIBER_MODEL) &&
                null != this.metadata.getProperty(IMessageService.KEY_LISTENER_MODEL))
        {
            throw new IllegalArgumentException("Message sink cannot support both subscriber and listener models simultaneously");
        }

        if (null != this.metadata.getProperty(IMessageService.KEY_SUBSCRIBER_MODEL))
        {
            hasModelBeenDefined = true;
            isSubscriberBasedModel = true;
        } else if (null != this.metadata.getProperty(IMessageService.KEY_LISTENER_MODEL))
        {
            hasModelBeenDefined = true;
            isSubscriberBasedModel = false;
        }
    }

    @Override
    public final void setGrammarInspectorFactory(IGrammarInspectorFactory factory)
    {
        Assert.parameterNotNull(factory, "Parameter 'factory' must not be null");
        this.grammarInspectorFactory = factory;
    }

    @Override
    public final IGrammarInspector createGrammarInspector(IMessageBeanDescriptor messageBeanDescriptor)
    {
        if (null != grammarInspectorFactory)
        {
            return grammarInspectorFactory.createInspector(messageBeanDescriptor);
        } else
        {
            return new DefaultGrammarInspector(messageBeanDescriptor);
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final int countSubscribers()
    {
        return (!hasModelBeenDefined)
                ? 0
                : (isSubscriberBasedModel) ? subscribers.size() : -1;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public final int countProducers()
    {
        return producers.size();
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean hasModelBeenDefined()
    {
        return hasModelBeenDefined;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean hasRegisteredListener()
    {
        return (null != registeredListener);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean supportsSubscribers()
    {
        return isSubscriberBasedModel;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public boolean supportsRegisteredListener()
    {
        return !isSubscriberBasedModel;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Class<T> getMessageBeanClass()
    {
        return messageBeanClass;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public Properties getMetadata()
    {
        return metadata;
    }

    /**
     * Empty implementation.
     * <p/>
     * {@inheritDoc}
     */
    @Override
    public void onLifecycleEvent(LifecycleActivityEvent event)
    {

    }

    /**
     * Empty implementation.
     * <p/>
     * {@inheritDoc}
     */
    @Override
    public void onLifecycleEvent(LifecycleServiceIntentionEvent event)
    {

    }

    /**
     * Convenience method that invokes {@link #createProducer(java.util.Properties)}, passing an empty,
     * non-null <code>Properties</code> object.
     *
     * @return Message sink producer instance
     */
    @Override
    public IMessageSinkProducer<T> createProducer()
    {
        return createProducer(new Properties());
    }

    /**
     * Convenience method that invokes {@link #createSubscriber(java.util.Properties)}, passing an empty,
     * non-null <code>Properties</code> object.
     *
     * @return Message sink subscriber instance
     */
    @Override
    public IMessageSinkSubscriber<T> createSubscriber()
    {
        return createSubscriber(new Properties());
    }

    /**
     * Convenience method that invokes {@link #createSubscriber(java.util.Properties)}, passing an empty,
     * non-null <code>Properties</code> object.
     *
     * @return Message sink subscriber instance
     */
    @Override
    public IMessageSinkSubscriber<T> createSubscriber(IMessageBeanDescriptor messageBeanDescriptor, String filter) throws IllegalExpressionException
    {
        return createSubscriber(new Properties(), messageBeanDescriptor, filter);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public IMessageSinkSubscriber<T> createSubscriber(Properties metadata) throws IllegalStateException
    {
        synchronized (lock)
        {
            if (!hasModelBeenDefined)
            {
                hasModelBeenDefined = true;
                isSubscriberBasedModel = true;

                logger.config("MessageSink " + Namespace.toCanonical(this) + " has settled on a multi-subscriber model");
            } else
            {
                if (!isSubscriberBasedModel)
                {
                    throw new IllegalStateException("Message sink has already settled on a sole registered listener model and cannot create message sink subscribers");
                } else
                {
                    logger.config("A new subscriber is being created in message sink " + Namespace.toCanonical(this));
                }
            }
        }

        return processCreateSubscriberRequest(metadata);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public IMessageSinkSubscriber<T> createSubscriber(Properties metadata, IMessageBeanDescriptor messageBeanDescriptor, String filter) throws IllegalStateException, IllegalExpressionException
    {
        Assert.parameterNotNull(messageBeanDescriptor, "Parameter 'messageBeanDescriptor' must not be null");
        Assert.parameterNotNull(filter, "Parameter 'filter' must not be null");

        synchronized (lock)
        {
            if (!hasModelBeenDefined)
            {
                hasModelBeenDefined = true;
                isSubscriberBasedModel = true;

                logger.config("MessageSink " + Namespace.toCanonical(this) + " has settled on a multi-subscriber model");
            } else
            {
                if (!isSubscriberBasedModel)
                {
                    throw new IllegalStateException("Message sink has already settled on a sole registered listener model and cannot create message sink subscribers");
                } else
                {
                    logger.config("A new subscriber is being created in message sink " + Namespace.toCanonical(this));
                }
            }
        }

        return processCreateSubscriberRequest(metadata, messageBeanDescriptor, filter);
    }

    /**
     * Thread-safe mechanism for concrete classes to implement a message sink subscriber. The base class implementations
     * of {@link #createSubscriber()} and {@link #createSubscriber(java.util.Properties)} have automatically enforced
     * model logic on behalf of the concrete implementation.
     *
     * @param metadata              Metadata describing the type of subscriber being requested. Specific metadata capabilities
     *                              must be documented by the concrete message service provider.
     * @param messageBeanDescriptor Message bean descriptor used by the filter
     * @param filter                Expression filter used to determine which messages are forwarded to the subscriber
     * @return Non-null message sink
     * @throws IllegalArgumentException when the concrete message service provider cannot honor the creation as defined
     *                                  by the metadata properties.
     */
    protected abstract IMessageSinkSubscriber<T> processCreateSubscriberRequest(Properties metadata, IMessageBeanDescriptor messageBeanDescriptor, String filter)
            throws IllegalArgumentException, IllegalExpressionException;

    /**
     * Thread-safe mechanism for concrete classes to implement a message sink subscriber. The base class implementations
     * of {@link #createSubscriber()} and {@link #createSubscriber(java.util.Properties)} have automatically enforced
     * model logic on behalf of the concrete implementation.
     *
     * @param metadata Metadata describing the type of subscriber being requested. Specific metadata capabilities
     *                 must be documented by the concrete message service provider.
     * @return Non-null message sink
     * @throws IllegalArgumentException when the concrete message service provider cannot honor the creation as defined
     *                                  by the metadata properties.
     */
    protected abstract IMessageSinkSubscriber<T> processCreateSubscriberRequest(Properties metadata)
            throws IllegalArgumentException;


    @Override
    public void setMessageSinkListener(IMessageSinkListener<T> messageSinkListener, IMessageBeanDescriptor messageBeanDescriptor, String filter) throws IllegalExpressionException
    {
        Assert.parameterNotNull(messageSinkListener, "Parameter 'messageSinkListener' must not be null");
        Assert.parameterNotNull(messageBeanDescriptor, "Parameter 'messageBeanDescriptor' must not be null");
        Assert.parameterNotNull(filter, "Parameter 'filter' must not be null");

        IGrammarInspector inspector = grammarInspectorFactory.createInspector(messageBeanDescriptor);
        inspector.inspect(filter);

        if (inspector.isValidExpression())
        {
            if (inspector.getParameterCount() > 0)
            {
                throw new IllegalExpressionException("Parameter substitution is not supported by this message sink implementation");
            }

            synchronized (lock)
            {
                if (!hasModelBeenDefined)
                {
                    hasModelBeenDefined = true;
                    isSubscriberBasedModel = false;
                    registeredListener = messageSinkListener;

                    logger.config("MessageSinkListener has been established in message sink " + Namespace.toCanonical(this));
                    logger.config("MessageSink " + Namespace.toCanonical(this) + " has settled on a sole registered listener model");
                } else
                {
                    if (isSubscriberBasedModel)
                    {
                        throw new IllegalStateException("Message sink has already settled on a subscriber model and cannot accept a message sink listener");
                    } else
                    {
                        if (null != messageSinkListener)
                        {
                            throw new IllegalStateException("A listener has already been registered with this message sink; it is not possible to dynamically replace the message sink listener once registered");
                        } else
                        {
                            logger.config("MessageSinkListener has been established in message sink " + Namespace.toCanonical(this));
                            registeredListener = messageSinkListener;
                        }
                    }
                }
            }
        } else
        {
            StringBuilder buf = new StringBuilder("\n\nExpression Errors: ");

            for(String err: inspector.errors())
            {
                buf.append(err).append("\n");
            }

            buf.append("\n\n");

            throw new IllegalExpressionException("Illegal filter expression" + buf.toString());
        }
    }


    /**
     * {@inheritDoc}
     */
    @Override
    public void setMessageSinkListener(IMessageSinkListener<T> messageSinkListener) throws IllegalStateException
    {
        Assert.parameterNotNull(messageSinkListener, "Parameter 'messageSinkListener' must not be null");

        synchronized (lock)
        {
            if (!hasModelBeenDefined)
            {
                hasModelBeenDefined = true;
                isSubscriberBasedModel = false;
                registeredListener = messageSinkListener;

                logger.config("MessageSinkListener has been established in message sink " + Namespace.toCanonical(this));
                logger.config("MessageSink " + Namespace.toCanonical(this) + " has settled on a sole registered listener model");
            } else
            {
                if (isSubscriberBasedModel)
                {
                    throw new IllegalStateException("Message sink has already settled on a subscriber model and cannot accept a message sink listener");
                } else
                {
                    if (null != messageSinkListener)
                    {
                        throw new IllegalStateException("A listener has already been registered with this message sink; it is not possible to dynamically replace the message sink listener once registered");
                    } else
                    {
                        logger.config("MessageSinkListener has been established in message sink " + Namespace.toCanonical(this));
                        registeredListener = messageSinkListener;
                    }
                }
            }
        }
    }
}
